package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_Transform {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStream")
    val ssc = new StreamingContext(sparkConf, Seconds(3) )
    ssc.sparkContext.setLogLevel("ERROR")
    //监听端口号 获取数据
    var lines = ssc.socketTextStream("localhost",9999)
    /*
    transform方法可以将底层RDD获取后进行操作
    当DStream功能不完善的时候，需要使用到RDD进行操作时可以使用该方法
     */
     val wordDS:DStream[(String,Int)]  = lines.transform(rdd=>{

      val words: RDD[String] = rdd.flatMap(_.split(" "))
      val wordOne:RDD[ (String,Int) ] = words.map((_, 1))
      val resWord: RDD[(String, Int)] = wordOne.reduceByKey(_ + _)

      resWord
    } )

    wordDS.print()

    ssc.start()
    ssc.awaitTermination()

  }

}
